package com.demo.base.redis.redisson.delayQueue;

import cn.hutool.core.date.DateTime;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @author 曳戈泰尔
 * @version 1.0
 * @description 消息消费端
 * @date 2023/3/30 11:41 PM
 */
@Component
@Slf4j
public class QueueRunner implements CommandLineRunner {

    /** 注入 RedissonClient */
    private RedissonClient redissonClient;

    @Autowired
    public void setRedissonClient(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    /** 注入 线程池 */
    private Executor threadPoolExecutor;

    @Autowired
    public void setThreadPoolExecutor(Executor threadPoolExecutor) {
        this.threadPoolExecutor = threadPoolExecutor;
    }

    @Override
    public void run(String... args) {

        RBlockingQueue<String> blockingFairQueue =
                redissonClient.getBlockingQueue("orderDelayQueue");
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(null, 2, TimeUnit.MILLISECONDS);

        while (true) {

            // 取订单号
            String delayOrderId = null;
            try {
                delayOrderId = blockingFairQueue.take();
            } catch (Exception e) {
                continue;
            }
            if (StringUtils.isBlank(delayOrderId)) {
                continue;
            }

            log.info(
                    String.format(
                            DateTime.now().toString("yyyy-MM-dd HH:mm:ss")
                                    + "，延时队列收到："
                                    + delayOrderId));

            // 处理订单
            String finalDelayOrderId = delayOrderId;
            threadPoolExecutor.execute(
                    new Runnable() {
                        @Override
                        public void run() {

                            // TODO
                            // 执行业务逻辑
                        }
                    });
        }
    }
}
